#pragma once

#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/TcpClient.h>
#include "detail.hpp"
#include "fields.hpp"
#include "abstract.hpp"
#include "message.hpp"
#include <mutex>
#include <unordered_map>

namespace myrpc 
{
    class MuduoBuffer : public BaseBuffer 
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf)
            :_buf(buf)
        {}
        virtual size_t readableSize() override 
        {
            return _buf->readableBytes();
        }
        virtual int32_t peekInt32() override 
        {
            //muduo库是一个网络库，从缓冲区取出一个4字节整形，会进行网络字节序的转换
            return _buf->peekInt32();
        }
        virtual void retrieveInt32() override 
        {
            return _buf->retrieveInt32();
        }
        virtual int32_t readInt32() override 
        {
            return _buf->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override 
        {
            return _buf->retrieveAsString(len);
        }
    private:
        // 该指针指向的是connection对应的缓冲区，不能随意释放
        // 所以没有用智能指针，基于_buf实现内部的功能
        muduo::net::Buffer *_buf;
    };
    // 创建缓冲区的工厂
    class BufferFactory 
    {
    public:
        template<typename ...Args>
        static BaseBuffer::ptr create(Args&& ...args) 
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    // 应用层通信协议
    class LVProtocol : public BaseProtocol 
    {
    public:
        // |--Len--|--VALUE--|
        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        using ptr = std::shared_ptr<LVProtocol>;
        //判断缓冲区中的数据量是否足够一条消息的处理
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override 
        {
            if (buf->readableSize() < lenFieldsLength) 
            {
                return false;
            }
            int32_t total_len = buf->peekInt32();
            //DLOG("total_len:%d", total_len);
            if (buf->readableSize() < (total_len + lenFieldsLength)) 
            {
                return false;
            }
            return true;
        }
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override 
        {
            //当调用onMessage的时候，默认认为缓冲区中的数据足够一条完整的消息
            int32_t total_len = buf->readInt32(); //读取总长度
            MType mtype = (MType)buf->readInt32(); // 读取数据类型
            int32_t idlen = buf->readInt32(); // 读取id长度
            int32_t body_len = total_len - idlen - idlenFieldsLength - mtypeFieldsLength;
            std::string id = buf->retrieveAsString(idlen);
            std::string body = buf->retrieveAsString(body_len);
            msg = MessageFactory::create(mtype);
            if (msg.get() == nullptr) 
            {
                ELOG("消息类型错误，构造消息对象失败！");
                return false;
            }
            bool ret = msg->unserialize(body);
            if (ret == false) 
            {
                ELOG("消息正文反序列化失败！");
                return false;
            }
            msg->setId(id);
            msg->setMType(mtype);
            return true;
        }
        virtual std::string serialize(const BaseMessage::ptr &msg) override 
        {
            // |--Len--|--mtype--|--idlen--|--id--|--body--|
            std::string body = msg->serialize();
            std::string id = msg->rid();
            auto mtype = htonl((int32_t)msg->mtype());
            int32_t idlen = htonl(id.size());
            int32_t h_total_len = mtypeFieldsLength + idlenFieldsLength + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);
            //DLOG("h_total_len:%d", h_total_len);
            std::string result;
            result.reserve(h_total_len);    // 提前开好空间，提高效率
            result.append((char*)&n_total_len, lenFieldsLength);
            result.append((char*)&mtype, mtypeFieldsLength);
            result.append((char*)&idlen, idlenFieldsLength);
            result.append(id);
            result.append(body);
            return result;
        }
    private:
        const size_t lenFieldsLength = 4;
        const size_t mtypeFieldsLength = 4;
        const size_t idlenFieldsLength = 4;
    };
    class ProtocolFactory 
    {
    public:
        template<typename ...Args>
        static BaseProtocol::ptr create(Args&& ...args) 
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection 
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &protocol) 
            : _protocol(protocol)
            , _conn(conn) 
            {}
        virtual void send(const BaseMessage::ptr &msg) override 
        {
            std::string body = _protocol->serialize(msg);   // 通过应用层协议序列化
            _conn->send(body);      // 通过网络库发送
        }
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        virtual bool connected() override 
        {
            _conn->connected();
            return true;
        }
    private:
        BaseProtocol::ptr _protocol;        // 依赖的协议
        muduo::net::TcpConnectionPtr _conn; // 依赖的网络库的通信连接
    };
    class ConnectionFactory 
    {
    public:
        template<typename ...Args>
        static BaseConnection::ptr create(Args&& ...args) 
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };


    /**
     * 1、这个 MuduoServer 其实就相当于是muduo的一个上层封装，
     * 
     * 然后呢，他是通过注册回调函数的方式进行事件处理的，这个能理解把，
     * 
     * 现在理一下噢，
     * 
     * 当一个连接断开的时候，会调用我们注册到muduo库中的onConnection这个函数，
     *      onConnection --》 这个函数中，通过conn->connected()的判断，发现是断开连接的操作，当执行完这个操作的之后，也需要通知上层进行一些资源的处理，然后调用上层注册的回调函数
     *          _cb_close -》 这个是一个回调函数，其实是调用到了RegistryServer::onConnShutdown这个函数中
     *          RegistryServer::onConnShutdown--》 处理完上层的资源释放之后，在一层一层退回来
     *          RegistryServer::onConnShutdown--》 end
     *      onConnection --》end
     * 继续回调muduo库的内部继续处理epoll事件， 
     * 
     * 到目前为止，有哪里有问题吗 上层是那个上层  基类中的server的回调函数做什么了
     * 
     * RegistryServer {
     *      // 这个类中有一个成员是
     *      BaseServer{
     *          // 这个类被继承
     *          MuduoServer{
     *              // 这个类中封装了muduo库
     *              muduo::net::TcpServer{
     *                  // muduo库内部的逻辑中，连接关闭了就会调用注册的回调函数，一层一层上执行能理解吗可以
     *              }
     *          }
     *      }
     * }
     *      还有哪里不理解呢没了
     *      好的，咱们之前在linux网络阶段应该是写过reactor的，那么应该根据自己的理解对这个muduo库的内部有一个大致的猜测，他的回调方法的这种处理，你自己也可以模拟以下，实现一个网络库，就能理解了，
     *      那我就先断开了噢，等下老师，那个reactor这个课程我没看，我为了赶时间做项目 这个就滤过了，按你的意思这个课还是必须得看的  是嘛？？
     *      可以对这块的只是有一个全面的总结把，reactor，事件驱动型 就是 我还得必须把那个课程给砍了
     *      你是赶着秋招是把，如果是这样的话，其实现在就可以准备找工作了，找共同工作的同时继续学习，扩展和完善自己的知识体系
     *      我是实习老师，那也得继续学习，不能因为学完了就不去碰之前的东西了，每次回过头去看一下，都会有新的体验和感悟，反复学习，更新迭代自己的知识体系，持续学习，
     */
    class MuduoServer : public BaseServer 
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port) 
            : _server(&_baseloop
                , muduo::net::InetAddress("0.0.0.0", port)
                , "MuduoServer"
                , muduo::net::TcpServer::kReusePort)
            , _protocol(ProtocolFactory::create())   // 构造_protocol对象
            {}
        virtual void start() 
        {
            // 2、这里这个函数setConnectionCallback，就是向muduo库内部注册一个函数，
            // 注册的这个函数在获取连接或者连接断开的时候，就会被调用，
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            // 4、这个函数被注册到了muduo库内部，每当一个连接，读取到内容到connection的输入缓冲区中的时候，就会调用这个注册的回调函数
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _server.start();//先开始监听
            _baseloop.loop();//开始死循环事件监控
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn) 
        {
            //3、也就是当一个连接建立或者断开的时候，就会调用这个onConnection函数，因为这个函数被注册到了muduo库内部
            //  这个能理解吗--能
            // // 用conn->connected()这个函数的返回值来判断是建立连接还是断开连接 这个能理解吗--
            // 这个函数是muduo库里面实现的不用我们管吧，对的
            // 只知道用法就可以？对的
            // 连接没有断开
            if (conn->connected())  
            {
                std::cout << "连接建立！\n";
                // 构建MuduoConnection对象
                auto muduo_conn = ConnectionFactory::create(conn, _protocol);
                {
                    // 加锁     为什么？这里
                    // 因为这个程序是在多线程下执行的，即使你在你的逻辑中没有创建其他线程，
                    // 但是你创建了muduo库示例，muduo库是有自己内部线程的，
                    // 如果你使用自己的线程来执行muduo库的start函数的话，虽然是单线程，但是这样去写也是可以的，
                    // 防止之后要对这个程序进行改动，要设计成多线程的模式去执行，那个时候，再去排查哪里的函数是线程下不安全的，那样的工作量就太大了，能理解吗能
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conn));
                    // 出作用域解锁
                }
                // 5、所以代码执行到这里，就是说，有一个新的连接建立了，那么该怎么通知上层调用呢，
                // 就是通过上层进行回调函数注册的方式，
                if (_cb_connection) _cb_connection(muduo_conn);
            }
            else // 连接断开
            {
                std::cout << "连接断开！\n";
                BaseConnection::ptr muduo_conn;
                {
                    // 加锁 
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end()) 
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                // 6、代码执行到这里，就是说明有连接断开了，怎么通知上层调用呢，就是通过注册的回调函数的方式
                if (_cb_close) _cb_close(muduo_conn);
            }
        }
        // 收到消息之后的回调函数
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            // 4、也就是会调用到这里来，在进行请求的处理
            DLOG("连接有数据到来，开始处理！");
            auto base_buf = BufferFactory::create(buf);
            while(1) 
            {
                if (_protocol->canProcessed(base_buf) == false) 
                {
                    // 缓冲区数据大于缓冲区的最大容量，直接关闭连接。数据不足
                    if (base_buf->readableSize() > maxDataSize) 
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大！");
                        return ;
                    }
                    //DLOG("数据量不足！");
                    break;
                }
                // 走到这里说明数据能够处理
                // DLOG("缓冲区中数据可处理！");
                BaseMessage::ptr msg;
                // onMessage会构造一个消息类型对象
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false) 
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误！");
                    return ;
                }
                //DLOG("消息反序列化成功！")
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end()) 
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                //DLOG("调用回调函数进行消息处理！");
                if (_cb_message) _cb_message(base_conn, msg);
            }
        }
    private:
        const size_t maxDataSize = (1 << 16);   // 缓冲区数据的最大容量
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;        // 用于事件循环监控
        muduo::net::TcpServer _server;          // 用于通信连接的管理
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };

    class ServerFactory 
    {
    public:
        template<typename ...Args>
        static BaseServer::ptr create(Args&& ...args) 
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient 
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &sip, int sport)
            : _protocol(ProtocolFactory::create())
            , _baseloop(_loopthread.startLoop())
            , _downlatch(1)
            , _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient")
            {}
        virtual void connect() override 
        {
            DLOG("设置回调函数，连接服务器");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            //设置连接消息的回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            //连接服务器
            _client.connect();
            _downlatch.wait();
            DLOG("连接服务器成功！");
        }
        virtual void shutdown() override 
        {
            return _client.disconnect();
        }
        virtual bool send(const BaseMessage::ptr &msg) override 
        {
            if (connected() == false) 
            {
                ELOG("连接已断开！");
                return false;
            }
            _conn->send(msg);
            return true;
        }
        virtual BaseConnection::ptr connection() override 
        {
            return _conn;
        }
        virtual bool connected() 
        {
            return (_conn && _conn->connected());
        }
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn) 
        {
            if (conn->connected()) 
            {
                std::cout << "连接建立！\n";
                _downlatch.countDown();//计数--，为0时唤醒阻塞
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else 
            {
                std::cout << "连接断开！\n";
                _conn.reset();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            DLOG("连接有数据到来，开始处理！");
            auto base_buf = BufferFactory::create(buf);
            while(1) {
                if (_protocol->canProcessed(base_buf) == false) 
                {
                    //数据不足
                    if (base_buf->readableSize() > maxDataSize) 
                    {
                        conn->shutdown();
                        ELOG("缓冲区中数据过大！");
                        return ;
                    }
                    //DLOG("数据量不足！");
                    break;
                }
                //DLOG("缓冲区中数据可处理！");
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(base_buf, msg);
                if (ret == false) 
                {
                    conn->shutdown();
                    ELOG("缓冲区中数据错误！");
                    return ;
                }
                //DLOG("缓冲区中数据解析完毕，调用回调函数进行处理！");
                if (_cb_message) _cb_message(_conn, msg);
            }
        }
    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };
    
    class ClientFactory 
    {
    public:
        template<typename ...Args>
        static BaseClient::ptr create(Args&& ...args) 
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}